package eventbk

import (
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"runtime/debug"
	"sync"
	"time"
	"yunj/pkg/global"
	"yunj/pkg/yunj/util"
)

// 监听者
type Listener struct {
	IsSync  bool // 是否同步执行
	Handler any  // 监听处理函数
}

// 订阅者结果
type SubscriberResult struct {
	Index      int    `json:"index"`       // 索引
	Error      error  `json:"-"`           // 错误信息
	ErrorStack string `json:"error_stack"` // 错误栈
}

// 实现json.Marshaler接口，格式化输出error
func (r SubscriberResult) MarshalJSON() ([]byte, error) {
	type Alias SubscriberResult // 创建别名避免递归调用

	var errMsg string
	if r.Error != nil {
		errMsg = r.Error.Error()
	}

	return json.Marshal(&struct {
		Error string `json:"error"`
		Alias
	}{
		Error: errMsg,
		Alias: (Alias)(r),
	})
}

// 处理结果
type HandleResult struct {
	StartTime         time.Time          `json:"-"`                  // 开始时间
	EndTime           time.Time          `json:"-"`                  // 截至时间
	Latency           uint64             `json:"latency"`            // 耗时（单位毫秒）
	SubscriberResults []SubscriberResult `json:"subscriber_results"` // 订阅者处理结果
	Error             error              `json:"-"`                  // 错误信息
}

// 实现json.Marshaler接口，格式化输出error
func (r HandleResult) MarshalJSON() ([]byte, error) {
	type Alias HandleResult // 创建别名避免递归调用

	var errMsg string
	if r.Error != nil {
		errMsg = r.Error.Error()
	}

	return json.Marshal(&struct {
		StartTime string `json:"start_time"`
		EndTime   string `json:"end_time"`
		Error     string `json:"error"`
		Alias
	}{
		StartTime: r.StartTime.Format("2006-01-02 15:04:05.000"),
		EndTime:   r.EndTime.Format("2006-01-02 15:04:05.000"),
		Error:     errMsg,
		Alias:     (Alias)(r),
	})
}

var subscribes = map[string][]*Listener{}
var mu sync.RWMutex

// 订阅事件
func Subscribe(name string, listeners ...*Listener) {
	mu.Lock()
	defer mu.Unlock()
	subscribes[name] = append(subscribes[name], listeners...)
}

// 发布事件
// name 事件名称
func Publish(name string, args ...interface{}) (handleResult *HandleResult) {
	mu.RLock()
	defer mu.RUnlock()

	defer func() {
		if r := recover(); r != nil {
			fmt.Printf("发布事件：panic: %v\n", r)
		}
	}()

	listeners, exists := subscribes[name]
	fmt.Printf("发布事件：%s|%v\r\n", name, len(listeners))
	if !exists || len(listeners) <= 0 {
		global.Logger.Error(fmt.Errorf("事件[%s]未订阅！", name))
		return
	}

	startTime := time.Now()
	// 参数准备
	argValues, isRefType := handlePublishArgs(args...)

	// 异步
	var asyncWg sync.WaitGroup
	var asyncResultChan = make(chan SubscriberResult)
	var asyncResultChanMutex sync.Mutex
	asyncNum := 0
	// 同步
	var syncResultChan = make(chan SubscriberResult)
	var syncResultChanMutex sync.Mutex
	syncNum := 0

	// 协程执行订阅者方法
	for idx, listener := range listeners {
		handlerArgs := make([]interface{}, len(args))
		handlerArgValues := make([]reflect.Value, len(args))
		copy(handlerArgs, args)
		for i := range args {
			if !isRefType[i] {
				// 值类型参数需要重新反射（避免并发修改）
				handlerArgValues[i] = reflect.ValueOf(handlerArgs[i])
			} else {
				// 引用类型可安全共享
				handlerArgValues[i] = argValues[i]
			}
		}
		fmt.Printf("[%d]发布事件IsSync：%s|%v\r\n", idx, name, listener.IsSync)
		if listener.IsSync {
			syncNum += 1
			// 同步监听器也放在单独的goroutine中执行，避免阻塞其他监听器
			go execListenerHandler(idx, listener, handlerArgValues, syncResultChan, &syncResultChanMutex, args...)
		} else {
			// 异步
			asyncNum += 1
			asyncWg.Add(1)
			go func() {
				defer asyncWg.Done()
				execListenerHandler(idx, listener, handlerArgValues, asyncResultChan, &asyncResultChanMutex, args...)
			}()
		}
	}
	fmt.Printf("同步数量：%v\r\n", syncNum)

	if asyncNum > 0 {
		fmt.Printf("异步数量：%v\r\n", asyncNum)
		// 处理订阅者异步处理结果
		go func() {
			asyncWg.Wait()
			handlePublishResult(name, startTime, asyncNum, asyncResultChan)
		}()
	}
	if syncNum > 0 {
		fmt.Printf("同步数量：%v\r\n", syncNum)
		// 处理订阅者同步处理结果
		handleResult = handlePublishResult(name, startTime, syncNum, syncResultChan)
	}
	return

}

// 处理事件发布结果
func handlePublishResult(name string, startTime time.Time, resultNum int, resultChan chan SubscriberResult) *HandleResult {
	handleResult := &HandleResult{
		StartTime:         startTime,
		SubscriberResults: make([]SubscriberResult, resultNum),
	}
	errMsg := ""
	for i := 0; i < resultNum; i++ {
		subscriberResult := <-resultChan
		handleResult.SubscriberResults[subscriberResult.Index] = subscriberResult
		if subscriberResult.Error != nil {
			errMsg += fmt.Sprintf("订阅者[%d]%v；", subscriberResult.Index, subscriberResult.Error)
		}
	}
	close(resultChan)
	handleResult.EndTime = time.Now()
	handleResult.Latency = util.MsTimeDiff(startTime, handleResult.EndTime)
	if errMsg != "" {
		handleResult.Error = errors.New(errMsg)
	}
	global.Logger.WithFields(map[string]any{
		"handleResult": handleResult,
	}).Info(fmt.Sprintf(
		"事件 %s 开始：%s 截至：%s 耗时：%dms",
		name,
		handleResult.StartTime.Format("2006-01-02 15:04:05.000"),
		handleResult.EndTime.Format("2006-01-02 15:04:05.000"),
		handleResult.Latency,
	))
	return handleResult
}

// 发布参数准备
func handlePublishArgs(args ...interface{}) (argValues []reflect.Value, isRefType []bool) {
	argValues = make([]reflect.Value, len(args))
	isRefType = make([]bool, len(args))
	for i, arg := range args {
		if arg == nil {
			argValues[i] = reflect.Zero(nil)
		} else {
			argValues[i] = reflect.ValueOf(arg)
			k := argValues[i].Kind()
			isRefType[i] = k == reflect.Ptr || k == reflect.Slice || k == reflect.Map || k == reflect.Chan || k == reflect.Func
		}
	}
	return
}

// 执行订阅监听者处理函数
func execListenerHandler(idx int, listener *Listener, handlerArgValues []reflect.Value, resultChan chan SubscriberResult, resultChanMutex *sync.Mutex, args ...any) {
	defer func() {
		var err error
		if r := recover(); r != nil {
			if rErr, ok := r.(error); ok {
				err = rErr
			} else {
				err = fmt.Errorf("%v", r)
			}
			debug.PrintStack()
		}
		recordSubscriberResult(resultChan, resultChanMutex, idx, err)
	}()
	// 检查参数
	fmt.Printf("[%d]111：\r\n", idx)
	if err := util.CheckFuncArgs(listener.Handler, args...); err != nil {
		fmt.Printf("[%d]222：%v\r\n", idx, err)
		recordSubscriberResult(resultChan, resultChanMutex, idx, err)
		return
	}
	// 检查返回结果
	fmt.Printf("[%d]333：\r\n", idx)
	if ok := util.IsFuncReturnError(listener.Handler); !ok {
		fmt.Printf("[%d]444：%v\r\n", idx, ok)
		recordSubscriberResult(resultChan, resultChanMutex, idx, errors.New("函数需返回error类型数据"))
		return
	}
	// 执行
	fmt.Printf("[%d]555：\r\n", idx)
	res := reflect.ValueOf(listener.Handler).Call(handlerArgValues)
	fmt.Printf("[%d]666：%v\r\n", idx, res)
	if !res[0].IsNil() {
		recordSubscriberResult(resultChan, resultChanMutex, idx, res[0].Interface().(error))
		return
	}
	// 记录成功
	fmt.Printf("[%d]777：%v\r\n", idx, res)
	recordSubscriberResult(resultChan, resultChanMutex, idx, nil)
}

// 记录订阅者执行结果
func recordSubscriberResult(resultChan chan SubscriberResult, resultChanMutex *sync.Mutex, index int, err error) {
	if resultChanMutex != nil {
		resultChanMutex.Lock()
		defer resultChanMutex.Unlock()
	}
	result := SubscriberResult{Index: index}
	if err != nil {
		result.Error = err
		result.ErrorStack = util.FormatError(err)
	}
	resultChan <- result
}
